package net.bwie.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.guanjuntao.util.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author: 🌹帅的一塌糊涂🌹
 * @description:
 * @params: 使用状态编程,对is_new新老用户字段进行校验和修改
 * @return:
 * @date: 2025/5/18 19:30
 */
public class AdjustIsNewPrcessFuncion extends KeyedProcessFunction<String, String, String> {
    // TODO: 2025/5/18 声明状态
    private  transient ValueState<String> firstDateState = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        // TODO: 2025/5/18 初始化状态
       firstDateState = getRuntimeContext().getState(
               new ValueStateDescriptor<String>("first-date", String.class));
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        // TODO: 2025/5/18 1.获取状态
        String firstDate = firstDateState.value();
        // TODO: 2025/5/18 2.解析字段
        JSONObject jsonObject = JSON.parseObject(value);
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        Long tsValue = jsonObject.getLong("ts");

        // TODO: 2025/5/18 3.判断状态
        if ("1".equals(isNewValue)){
            if (null==firstDate){
                // TODO: 2025/5/18 更新状态 
                // TODO: 2025/5/18 此时的is_new是正确的不做修改
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            }else {
                // TODO: 2025/5/18 状态日期(首次日期日期) 与当前数据日期一致,不做处理
                if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                    // TODO: 2025/5/18 不一致的日期把用户状态改为老用户"0"
                    jsonObject.getJSONObject("common").put("is_new","0");
                }
            }
        }else {
            // TODO: 2025/5/18 如果用户在filnk上线应用之前就早早登录,状态就有可能为空
            if(null == firstDate){
                // TODO: 2025/5/18 可以把日期设置为昨天的日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue-86400000, DateTimeUtil.DATE_FORMAT));
            }
        }

        out.collect(jsonObject.toJSONString());
    }
}
